package thread.wang_bao_ling.c15.temp;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.stream.IntStream;

public class Test {
    private ReentrantLock createChannelLock = new ReentrantLock();
    Map<String, BlockingQueue<Runnable>> queueMap = new HashMap<>();
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    private Map<String, NioSocketChannel> sessionChannelMap = new ConcurrentHashMap<>();

    public void transport(String sessionKey){
        NioSocketChannel proxyChannel =sessionChannelMap.get(sessionKey);
        if (proxyChannel == null) {
            try{
                if(createChannelLock.tryLock(5, TimeUnit.SECONDS)){
                    proxyChannel =sessionChannelMap.get(sessionKey);
                    if (proxyChannel == null) {
                        if (queueMap.containsKey(sessionKey)) {
                            Thread.currentThread().sleep(1);
                            queueSafeControl(sessionKey,(q)->{
                                WaitWriteTask task=new WaitWriteTask(()->{
                                    NioSocketChannel channel = sessionChannelMap.get(sessionKey);
                                    channel.writeAndFlush(sessionKey+"队列写入");
                                });
                                System.out.println("task="+task);
                                q.add(task);
                            });
                        }else{
                            queueMap.put(sessionKey, new LinkedBlockingQueue<>());
                            createSession( sessionKey, (p -> {
                                if (p == null) {
                                    queueSafeControl(sessionKey,(q)->queueMap.remove(sessionKey));
                                }else{
                                    p.writeAndFlush(sessionKey+"普通写入");
                                    //执行队列中的写入任务
                                    queueSafeControl(sessionKey,(q)->{
                                        while (q.size() > 0) {
                                            try {
                                                q.take().run();
                                            } catch (InterruptedException e) {
                                                System.out.println("出队线程被中断{}"+e);
                                            }
                                        }
                                        queueMap.remove(sessionKey);
                                        System.out.println("--------queueMap.remove(sessionKey)-----size="+queueMap.size());
                                    });
                                }
                            }));
                        }
                    }else{
                        proxyChannel.writeAndFlush("存在写入");
                    }
                }else{
                    System.out.println("获取锁超时！");
                }
            } catch (InterruptedException e) {
                System.out.println("线程被中断！");
                queueMap.remove(sessionKey);//防止在刚刚put完线程被中断的情况
            } finally {
                createChannelLock.unlock();
            }
            return;
        }
        proxyChannel.writeAndFlush("存在写入");
    }


    /**
     * 在执行queue中的任务时如果还有入队任务就不再添加了，所以在做这些操作前先获取queue的锁
     * @param control
     */
    private void queueSafeControl(String sessionKey, Consumer<BlockingQueue<Runnable>> control) {
        System.out.println("-------------queueSafeControl---------");
        BlockingQueue<Runnable> queue=queueMap.get(sessionKey);
        if (queue != null) {
            //synchronized (queue) {
                queue=queueMap.get(sessionKey);
                if (queue != null) {
                    control.accept(queue);
                }
            //}
        }
    }

    public void createSession(String sessionKey, Consumer<NioSocketChannel> consumer) {
        executorService.submit(() -> {
            try {
                Thread.currentThread().sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            NioSocketChannel channel = new NioSocketChannel();
            sessionChannelMap.put(sessionKey, channel);
            System.out.println("创建sessionKey=" + sessionKey + ";channel=" + channel);
            consumer.accept(channel);
        });
    }

    public static void main(String[] args) throws InterruptedException {
        Test test = new Test();
        IntStream.range(0, 1).forEach(f -> {
            String uuid = UUID.randomUUID().toString();
            IntStream.range(0, 10).forEach(t -> {
                new Thread(() -> {
                    test.transport(uuid);
                }).start();
            });
        });

        Thread.currentThread().sleep(1000000);
    }

    @Override
    protected void finalize() throws Throwable {
        System.out.println("test 垃圾回收中");
    }

}
